FalconとSQLAlchemy によるWebサービスの実装
FalconとSQLAlchemyの連携
これまでは単純化するためにデータベースを使用せずにWebアプリケーションを作成していました。今度はデータベースを SQLAlchemy から使用するようにします。
RESTful Webサービスでデータベースを使用するときは、コミットするタイミングやセッションの有効時間の制御に留意するべきです。
REST API の性質上クライアントからのリクエスト間はセッションが有効であるべきですが、
クライアントは１つとは限らず同時に不特定多数からのアクセスがあることも考える必要があります。つまり、セッションのクロースするタイミングはアプリケーションの性能に影響することなります。
falcon からSQLAlchemy を利用するときに便利な拡張機能には次のものがあります。
falcon-sqlalchemy
ただし、2017年が最終更新日となっているので開発が止まっているようです。
ドキュメントもないためソースコードさえあればなんとかなるという猛者向けです。
それでも、ソースコードはfalcon でのプログラム開発にはとても参考になります。
falcon-sqla
falcon-sqla はfalcon-sqlalchemy より後発で、パッケージ名が既に使われているための苦肉の命名が想像できます。 SQLAlchemy のセッションマネージャをミドルウエアとして提供されて、セッション制御について開発者が気にする必要がなくなります。使い方はとてもシンプルです。
FalconとMarshamallowの連携
Falcon でのオブジェクトのシリアライズ/ディシリアライズは、デフォルトではJSONしかサポートしていません。Marshmallow を使用するとより複雑なスキームを定義することができ、ORMのモデルクラスのオブジェクトをシリアライズ/ディシリアライズできるようになります。
falcon-mashmallow
リソースクラスで クラス変数schema もしくは <METHOD>_schema に設定したスキーマを使って、リクエストデータのシリアライズ、レスポンスデータのディシリアライズをミドルウェアとして処理します。<METHOD> はHTTPプロトコルのメソッドです。例：get_schema
mashmallow-sqlalchemy
Falcon のRESTful API の自動生成
Falcon自体には RESTful API を自動生成するための機能は提供されていません。
次の拡張モジュールは SQLAlchemy ORM のモデルクラスを読み込んでHTTPメソッドに対応するデータベースCRUD操作のAPIを自動生成することができます。
２つともSQLAlchemy の セッション(session) ではなく、DBエンジン(db_engine)を与えます。セッション制御を気にする必要がありません。
Falcon-REST
定義が単純で使いやすいのですが、機能も単純で自由度がありません。
Falcon-aurocrud
Falcon-autocrud はCollectionResource と SingleResource の２つのリソースベースクラスを提供します。リソースクラスの種別を使い分けることができるため柔軟性がある反面、２つのリソースクラスを定義することになります。 リソースクラスにはデコレータ@authorize()で認証クラスを与えて、許可されたユーザだけがアクセスできるようなリソースを定義することもできます。
falcon-autoccrud はデフォルトではSQLAlchemyのモデルクラスから自動生成されます。
独自のスキーマを与える場合は、JSON Scheme の仕様にもとづいたスキーマを@request_schema() と @response_schema() で与えます。 余談
falcon-rest と falcon-autocrud のソースコードは falcon の学習にとても役立ちます。
一度目を通しておくことを強くお勧めします。
HTTPメソッドのマッピング
WebサービスTODOの仕様を少し変更しています。
理解が容易になる事を重視して認証によるWebサービスの保護は省いています。
table: APIとHTTPメソッド
HTTPメソッド URI アクション
タスクリソースは次の情報を持つものとします。
id：タスクを示す一意のID。Integer型。
title：タスクのタイトル。タスクについての短い説明。 String型。
description：タスクの詳細。タスクについての詳細な説明。 Text型。
done：タスクの完了状態。 Boolean型。
インストール
code: bash
$ pip install alembic falcon-marshmallow mashmallow-sqlalchemy
準備
アプリケーションのディレクトリを作成します。
code: bash
$ mkdir -p $HOME/falcon/todo_apiv2
$ cd $HOME/falcon/todo_apiv2
はじめに、 次のmodels.py を用意します。
code: python
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
db_uri = 'sqlite:///todo.db'
db_engine = sa.create_engine(db_uri)
Base = declarative_base()
class Task(Base):
__tablename__ = 'tasks'
id = sa.Column(sa.Integer, primary_key = True)
title = sa.Column(sa.String(128), index = True)
description = sa.Column(sa.Text(256))
done = sa.Column(sa.Boolean)
def __init__(self, title='new task', description='', done=False):
self.title = title
self.description = description
self.done = done
def __repr__(self):
return f'Task: {self.title}'
Alembic でデータベースを作成
モデルクラスを使って自分でデータベースを作成してもよいのですが、
ここではAlembic でデータベースを作成しましょう。
code: bash
$ alembic init migrations
alembic.ini で以下の修正します。
code: python
sqlalchemy.url = sqlite:///todo.db
migrations/env.py の target_metadata を修正します。
code: python
from models import Base
target_metadata = Base.metadata
これでデータベースを作成することができるようになります。
code: bash
$ export PYTHONPATH=.
$ alembic revision --autogenerate -m "db initialize"
$ alembic upgrade head
$ alembic stamp head
falcon-sqla による実装
falcon-rest や falcon-autocrud といった拡張モジュールを使ったAPI自動処理は簡単のアプリケーションを作成することができますが、自由度が低く細かい定義ができません。いろいろな処理を定義したいときは、こうしたパッケージのソースをアプリケーションに取り込んで修正するか、独自にプログラムすることになります。
falcon で SQLAlchemy を利用するときは、falcon-sqla が便利です。
falcon-sqla はセッションマネージャクラスManagerが提供されています。
セッションには session_scope() もしくは、req.context.sessionを使ってアクセスします。
使い方は次のようになります。
code: python
import falcon
import sqlalchemy as sa
from falcon_sqla import Manager
# ...
class SomeResource:
def __init__(self, db):
self.db = db
def on_get(self, req, resp):
result = req.context.session.query(model)
# ...
db_engine = sa.create_engine('dialect+driver://my/database')
api = falcon.API(middleware=[
Manager(db_engine)
])
api.add_route('/path/to/api', SomeResource(db_engine))
インストール
code: bash
$ pip install falcon-sqla
では、WebアプリケーションTODOを実装してみましょう。
models.py はこれまでと同じものを利用します。
GETメソッドでタスクの取得
GETメソッドは一覧を取得する場合と、タスクIDを指定してタスクを取得する２つのが処理があります。
code: python
import falcon
from falcon_sqla import Manager
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from models import Base, Task, db_engine
class TaskSchema(SQLAlchemyAutoSchema):
class Meta:
model = Task
ordered = True
class TaskBaseResource(object):
model = Task
schema = TaskSchema()
def __init__(self, db):
self.db = db
def retrieve(self, req, resp, id=None):
if id == None:
raw_data = req.context.session.query(self.model).all()
data = self.schema.dump(raw_data, many=True)
else:
raw_data = (
req.context.session.query(self.model)
.filter_by(id=id).first()
)
data = self.schema.dump(raw_data)
return data
def add(self, req, resp, id):
pass
def delete(self, req, resp, id=None):
pass
def patch(self, req, resp, id=None):
pass
class TaskListResource(TaskBaseResource):
def on_get(self, req, resp):
data = self.retrieve(req, resp)
resp.media = {"data": data }
def on_post(self, req, resp):
pass
class TaskResource(TaskBaseResource):
def on_get(self, req, resp, id):
try:
data = self.retrieve(req, resp, id)
resp.media = {"data": data }
except IndexError:
resp.media = {'result': 'Not Found'}
def on_delete(self, req, resp):
pass
def on_patch(self, req, resp):
pass
def error_handle_404(req, resp):
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not found'}
api = falcon.API(middleware=[
Manager(db_engine).middleware,
])
api.add_sink(error_handle_404)
api.add_route('/todo/api/v2.0/tasks', TaskListResource(db_engine))
api.add_route('/todo/api/v2.0/tasks/{id}', TaskResource(db_engine))
POSTメソッドでタスクを登録
POSTメソッドでタスクを登録できるようにします。
code: python
class TaskBaseResource(object):
model = Task
schema = TaskSchema()
# ...
def add(self, req, resp):
req_data = Task(title=req.media'title', req.context.session.add(req_data)
req.context.session.commit()
data = self.schema.dump(req_data)
return data
# ...
class TaskListResource(TaskBaseResource):
def on_get(self, req, resp):
data = self.retrieve(req, resp)
resp.media = {"data": data }
def on_post(self, req, resp):
data = self.add(req, resp)
resp.media = {"data": data }
DELETEメソッドでタスクを削除
code: python
class TaskBaseResource(object):
model = Task
schema = TaskSchema()
# ...
def delete(self, req, resp, id=None):
if id == None:
raise falcon.HTTPBadRequest()
else:
raw_data = (
req.context.session.query(self.model)
.filter_by(id=id).first()
)
if raw_data == None:
data = None
else:
req.context.session.delete(raw_data)
req.context.session.commit()
data = self.schema.dump(raw_data)
return data
# ...
class TaskResource(TaskBaseResource):
# ...
def on_delete(self, req, resp, id):
result = self.delete(req, resp, id)
if result != None:
resp.media = {'result': 'OK'}
else:
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not Found'}
PATCHメソッドでタスクを更新
code: python
class TaskBaseResource(object):
model = Task
schema = TaskSchema()
# ...
def patch(self, req, resp, id):
raw_data = ( req.context.session.query(self.model)
.filter_by(id=id).first()
)
if raw_data == None:
data = None
else:
req_data = Task(title=req.media'title', raw_data.title = req_data.title
raw_data.description = req_data.description
raw_data.done = req_data.done
req.context.session.commit()
data = self.schema.dump(raw_data)
return data
# ...
class TaskResource(TaskBaseResource):
# ...
def on_patch(self, req, resp, id):
result = self.patch(req, resp, id)
if result != None:
resp.media = {'result': 'OK'}
else:
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not Found'}
ソースコード
app.py は次のようになります。
code: python
import falcon
from falcon_sqla import Manager
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from models import Base, Task, db_engine
class TaskSchema(SQLAlchemyAutoSchema):
class Meta:
model = Task
ordered = True
class TaskBaseResource(object):
model = Task
schema = TaskSchema()
def __init__(self, db):
self.db = db
def retrieve(self, req, resp, id=None):
if id == None:
raw_data = req.context.session.query(self.model).all()
data = self.schema.dump(raw_data, many=True)
else:
raw_data = ( req.context.session.query(self.model)
.filter_by(id=id).first()
)
data = self.schema.dump(raw_data)
return data
def add(self, req, resp):
req_data = Task(title=req.media'title', req.context.session.add(req_data)
req.context.session.commit()
data = self.schema.dump(req_data)
return data
def delete(self, req, resp, id=None):
if id == None:
raise falcon.HTTPBadRequest()
else:
raw_data = ( req.context.session.query(self.model)
.filter_by(id=id).first()
)
if raw_data == None:
data = None
else:
req.context.session.delete(raw_data)
req.context.session.commit()
data = self.schema.dump(raw_data)
return data
def patch(self, req, resp, id):
raw_data = ( req.context.session.query(self.model)
.filter_by(id=id).first()
)
if raw_data == None:
data = None
else:
req_data = Task(title=req.media'title', raw_data.title = req_data.title
raw_data.description = req_data.description
raw_data.done = req_data.done
req.context.session.commit()
data = self.schema.dump(raw_data)
return data
class TaskListResource(TaskBaseResource):
def on_get(self, req, resp):
data = self.retrieve(req, resp)
resp.media = {"data": data }
def on_post(self, req, resp):
data = self.add(req, resp)
resp.media = {"data": data }
class TaskResource(TaskBaseResource):
def on_get(self, req, resp, id):
try:
data = self.retrieve(req, resp, id)
resp.media = {"data": data }
except IndexError:
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not Found'}
def on_delete(self, req, resp, id):
result = self.delete(req, resp, id)
if result != None:
resp.media = {'result': 'OK'}
else:
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not Found'}
def on_patch(self, req, resp, id):
result = self.patch(req, resp, id)
if result != None:
resp.media = {'result': 'OK'}
else:
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not Found'}
def error_handle_404(req, resp):
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not found'}
api = falcon.API(middleware=[
Manager(db_engine).middleware,
])
api.add_sink(error_handle_404)
api.add_route('/todo/api/v2.0/tasks', TaskListResource(db_engine))
api.add_route('/todo/api/v2.0/tasks/{id}', TaskResource(db_engine))
falcon-autocrud による実装
app.py は次のようにとても簡潔になります。
インストール
code: bash
$ pip install falcon-autocrud
CollectionResource もしくはSingleResourceを継承したリソースクラスを定義して、そのクラス変数としてモデルクラスを設定するだけです。
code: python
import falcon
from falcon_autocrud.resource import CollectionResource, SingleResource
from models import db_engine, Task
class TaskListResource(CollectionResource):
model = Task
class TaskResource(SingleResource):
model = Task
def error_handle_404(req, resp):
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not found'}
pi = falcon.API()
api.add_sink(error_handle_404)
api.add_route('/todo/api/v2.0/tasks', TaskListResource(db_engine))
api.add_route('/todo/api/v2.0/tasks/{id}', TaskResource(db_engine))
code: bash
{
"data": []
}
code: bash
{
"data": {
"id": 1,
"title": "Gymnastics",
"description": "Goto Anytime Fitness",
"done": false
}
}
falcon-rest による実装
こんどは、falcon-rest で実装してみましょう。
インストール
code: bash
$ pip install falcon-rest
falcon-rest は falcon にバージョンに追随できていない部分があり、次のようにパッチを当てておきます。
code: patch
$ diff -Nru resources.py.orig resources.py
--- resources.py.orig 2020-05-19 16:18:37.000000000 +0900
+++ resources.py 2020-05-19 16:40:30.000000000 +0900
@@ -51,7 +51,7 @@
raise exceptions.FalconRestException(
"No retrieve_serializer or serializer defined on {0}".format(self.__class__.__name__))
- data = serializer.dump(data).data
+ data = serializer.dump(data)
return data
@@ -70,7 +70,7 @@
raise exceptions.FalconRestException(
"No list_serializer or serializer defined on {0}".format(self.__class__.__name__))
- data = serializer.dump(data, many=True).data
+ data = serializer.dump(data, many=True)
return data
def on_post(self, request, response, *args, **kwargs):
@@ -101,7 +101,7 @@
raise exceptions.FalconRestException(
"No create_serializer or serializer defined on {0}".format(self.__class__.__name__))
- dataset = serializer.dump(dataset, many=True).data
+ dataset = serializer.dump(dataset, many=True)
return dataset
def on_patch(self, request, response, *args, **kwargs):
@@ -132,7 +132,7 @@
raise exceptions.FalconRestException(
"No patch_serializer or serializer defined on {0}".format(self.__class__.__name__))
- dataset = serializer.dump(dataset, many=True).data
+ dataset = serializer.dump(dataset, many=True)
return dataset
def on_delete(self, request, response, *args, **kwargs):
モデルクラスはそのまま利用できます。
app.py を次のように定義します。
falcon-autocrud と違い、継承するベースクラスは ModelResourceしかありません。
対応するHTTPプロトコルメソッドをallowed_methodsにリストで与えます。
code: python
import falcon
import falcon_json_middleware
from falcon_rest.resources import ModelResource
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from models import Base, Task, db_engine
class TaskSerializer(SQLAlchemyAutoSchema):
class Meta:
model = Task
class TaskListResource(ModelResource):
model = Task
serializer = TaskSerializer()
class TaskResource(ModelResource):
model = Task
serializer = TaskSerializer()
def error_handle_404(req, resp):
resp.status = falcon.HTTP_404
resp.media = {'result': 'Not found'}
api = falcon.API(middleware=[
falcon_json_middleware.Middleware(),
])
api.add_sink(error_handle_404)
api.add_route('/todo/api/v2.0/tasks', TaskListResource(db_engine))
api.add_route('/todo/api/v2.0/tasks/{id}', TaskResource(db_engine))
code: bash
{
"timestamp": "2020-05-19T16:23:22.649676",
"data": {
"title": "Gymnastics",
"id": 1,
"done": false,
"description": "Goto Anytime Fitness"
}
}
code: bash
{
"timestamp": "2020-05-19T16:34:48.124316",
"data": [
{
"description": "I Love IPA",
"title": "Drink Beers",
"done": false,
"id": 2
}
]
}
code: bash
{
"timestamp": "2020-05-19T16:35:04.086878",
"data": [
{
"description": "Goto Anytime Fitness",
"title": "Gymnastics",
"done": false,
"id": 1
},
{
"description": "I Love IPA",
"title": "Drink Beers",
"done": false,
"id": 2
}
]
}
code: bash
{
"timestamp": "2020-05-19T16:38:58.618024",
"data": {
"status": "deleted"
}
}
code: bash
{
"timestamp": "2020-05-19T16:39:15.505162",
"data": [
{
"description": "I Love IPA",
"title": "Drink Beers",
"done": false,
"id": 2
}
]
}